package com.lxb.demo3

import java.sql.{DriverManager, ResultSet}

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

/**
 * @author: albert
 * @date: 2021/5/26 20:31
 * @description:
 */
object TableApiTest {

  def main(args:Array[String]):Unit = {
    // 创建Stream环境和table环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 创建模拟数据
    val dataStream: DataStream[String] = env.fromElements("tom", "sam", "row")

    // 将dataStream读到table环境中
    val inputTable: Table = tableEnv.fromDataStream(dataStream)

    // 创建临时表
    tableEnv.createTemporaryView("InputData",inputTable)

    // 查询
    val resultTable: Table = tableEnv.sqlQuery("select * from InputData")

    // table转化为dataStream
    val resultData: DataStream[Row] = tableEnv.toDataStream(resultTable)

    resultData.print()

    env.execute()


  }

  def JDBCClickhouse(): Unit ={
    Class.forName("ru.yandex.clickhouse.ClickHouseDriver")
    var conn = DriverManager.getConnection("jdbc:clickhouse://192.168.100.107:8123?socket_timeout=300000",
      "default", "albert")
    var insertStmt = conn.prepareStatement("SELECT * from default.test t final where vin  = '1';")

    val set: ResultSet = insertStmt.executeQuery()

    while(set.next){
      val str: String = set.getString(1)
      print(str)
    }
  }
}
